package com.seari.movedata.service;

import com.seari.movedata.mapper.FirstInitDaoImpl;
import com.seari.movedata.utils.CommonUtils;
import com.seari.movedata.utils.JedisPoolUtils;
import com.seari.movedata.utils.ListUtils;
import com.seari.movedata.utils.LocalCacheUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import redis.clients.jedis.Jedis;
import redis.clients.jedis.Pipeline;

import java.util.ArrayList;
import java.util.List;
import java.util.Set;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.atomic.AtomicInteger;

public class FirstInitServiceImpl {

    private static Logger logger = LoggerFactory.getLogger(FirstInitServiceImpl.class);

    public static final int MAX_WORKER_COUNT = 10;
    public static final int MAX_SLICE_COUNT = 20000;
    public static BlockingQueue<List<String>> aiQueue = new LinkedBlockingQueue<>(200);
    public static BlockingQueue<List<String>> diQueue = new LinkedBlockingQueue<>(200);
    public static ExecutorService service = Executors.newCachedThreadPool();
    public static AtomicInteger workerCount = new AtomicInteger();

    Jedis jedis = JedisPoolUtils.getJedis();

    LocalCacheUtils localCacheUtils = new LocalCacheUtils();

    public void initLocalCacheAndPutDataToInfluxdb(List<String> aiKeyList, List<String> diKeyList) {

        initLocalCache(aiKeyList,diKeyList);
        logger.info("init localCache complete");

        if (!CommonUtils.objectIsNull(aiKeyList)) {
            for (int i = 0;i<aiKeyList.size();i++) {
                String aikeyStr = aiKeyList.get(i);
                Set<String> aiKeyset = jedis.hkeys(aikeyStr);
                if(!CommonUtils.objectIsNull(aiKeyset)) {
                    List<String> result = new ArrayList<String>(aiKeyset);
                    List<List<String>> listList = ListUtils.splitList(result, MAX_SLICE_COUNT);
                    for (List<String> list : listList) {
                        List<String> listStr = new ArrayList<String>();
                        listStr.addAll(list);
                        listStr.add(aikeyStr);
                        aiQueue.add(listStr);
                    }
                }else{
                    logger.info( "   "+aikeyStr+ " key  not find in redis");
                }
            }
        }

        if (!CommonUtils.objectIsNull(diKeyList)) {
            for (int j = 0;j<aiKeyList.size();j++) {
                String dikeyStr = diKeyList.get(j);
                Set<String> diKeyset = jedis.hkeys(dikeyStr);
                if(!CommonUtils.objectIsNull(diKeyset)) {
                    List<String> result = new ArrayList<String>(diKeyset);
                    List<List<String>> listList = ListUtils.splitList(result, MAX_SLICE_COUNT);
                    for (List<String> list : listList) {
                        List<String> listStr = new ArrayList<String>();
                        listStr.addAll(list);
                        listStr.add(dikeyStr);
                        diQueue.add(listStr);
                    }
                }else{
                    logger.info( " "+dikeyStr+ "  key not find in redis");
                }
            }
        }
        startWorker();
    }

    public void initLocalCache(List<String> aiKeyList, List<String> diKeyList){

        if (!CommonUtils.objectIsNull(aiKeyList)) {
            Pipeline pipeline = jedis.pipelined();
            for (int i = 0; i < aiKeyList.size(); i++) {
                pipeline.hgetAll(aiKeyList.get(i));
            }
            List<Object> aiResults = pipeline.syncAndReturnAll();
            for(int i = 0; i < aiKeyList.size(); i++){
                String keyStr = aiKeyList.get(i);
                Object valueStr = aiResults.get(i);
                localCacheUtils.putValue(keyStr,valueStr,-1);
            }
        }

        if (!CommonUtils.objectIsNull(aiKeyList)) {
            Pipeline pipeline = jedis.pipelined();
            for (int j = 0; j < diKeyList.size(); j++) {
                pipeline.hgetAll(diKeyList.get(j));
            }
            List<Object> diResults = pipeline.syncAndReturnAll();
            for(int j = 0; j < diKeyList.size(); j++){
                String keyStr = diKeyList.get(j);
                Object valueStr = diResults.get(j);
                localCacheUtils.putValue(keyStr,valueStr,-1);
            }
        }


    }

    public void startWorker(){
        while(workerCount.get() < MAX_WORKER_COUNT){
            FirstInitDaoImpl FirstInitDaoWorker = new FirstInitDaoImpl(aiQueue,diQueue);
            service.submit(FirstInitDaoWorker);
            int num = workerCount.incrementAndGet();
            logger.info("start worker:current work count is:" + num);
        }
    }
}
